package com.mrzhou.rabbit.consumer;

import com.mrzhou.core.tool.api.R;
import com.mrzhou.core.tool.utils.Func;
import com.mrzhou.core.tool.utils.JacksonUtil;
import com.mrzhou.core.tool.utils.StringUtil;
import com.mrzhou.data.support.RedisService;
import com.mrzhou.log.constant.LogConstant;
import com.mrzhou.log.entity.LogApi;
import com.mrzhou.log.feign.ILogFeignClient;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.locks.ReentrantLock;

@Component
@Slf4j
public class ApiLogConsumer {

    @Autowired
    private ILogFeignClient logFeignClient;
    @Autowired
    private RedisService redisService;

    /** RabbitMQ DeliveryTag Redis缓存空间 */
    final String MQ_DELIVERY_TAG_API_LOG = "MQ_DELIVERY_TAG:API_LOG";

    @RabbitListener(bindings = @QueueBinding(value = @Queue(value = LogConstant.LOG_QUEUE, durable = "true"),
            exchange = @Exchange(value = LogConstant.LOG_EXCHANGE, type = "topic"),
            key = LogConstant.LOG_QUEUE))
    public void receive(LogApi message, Message msg, Channel channel) throws IOException {
        long deliveryTag = msg.getMessageProperties().getDeliveryTag();
        try {
            if (redisService.hmExists(MQ_DELIVERY_TAG_API_LOG, Func.toStr(deliveryTag))) {
                log.warn(String.format("当前消息已经在处理中 {Redis_KEY: %s, Delivery_Tag: %s}", MQ_DELIVERY_TAG_API_LOG, deliveryTag));
                return;
            }
            // 将消息ID保存至Redis中
            redisService.hmSetKey(MQ_DELIVERY_TAG_API_LOG, Func.toStr(deliveryTag), deliveryTag);

            R<String> result = logFeignClient.apiSubmit(message);
            if (result.isSuccess()) {
                // 手动确认当前消息已被消费
                channel.basicAck(deliveryTag, false);
            } else {
                channel.basicNack(deliveryTag, false, true);
            }
        } catch (Exception e) {
            // 重新将消息入列
            channel.basicNack(deliveryTag, false, true);
        } finally {
            redisService.hmRemove(MQ_DELIVERY_TAG_API_LOG, Func.toStr(deliveryTag));
        }
    }

}
